package com.lry.basic.redis;

import com.lry.basic.metrics.spi.util.CollectionUtils;
import io.lettuce.core.RedisClient;
import io.lettuce.core.RedisFuture;
import io.lettuce.core.RedisURI;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.async.RedisAsyncCommands;
import io.lettuce.core.api.sync.RedisCommands;
import io.netty.buffer.UnpooledDirectByteBuf;
import org.aspectj.lang.annotation.Before;
import org.junit.jupiter.api.Test;
import redis.clients.jedis.*;
import redis.clients.util.JedisClusterCRC16;

import java.nio.channels.Channel;
import java.util.*;
import java.util.concurrent.ExecutionException;

/**
 * @author:刘仁有
 * @desc:
 * @email:953506233@qq.com
 * @data:2019/8/6
 */
public class RedisDemo1 {


    @Test
    public void dist() throws InterruptedException {
        Jedis jedis = RedisPool.getJedis();
        String requestId = UUID.randomUUID().toString();
        boolean f = RedisLock.tryGetDistributedLock(jedis,"k5",requestId,30000);
        System.out.println(f);

        f = RedisLock.tryGetDistributedLock(jedis,"k5",requestId,30000);
        System.out.println(f);

        //锁的过期时间为30s，休眠50s，测试看门狗会不会自己延迟过期时间
        Thread.sleep(50000);

        f = RedisLock.releaseDistributedLock(jedis,"k5",requestId);
        System.out.println(f);

        RedisPool.release(jedis);

    }





    @Test
    public void reentrantLock(){
        Jedis jedis = RedisPool.getJedis();
        String requestId = UUID.randomUUID().toString();
        ReentrantRedisLock reentrantRedisLock = new ReentrantRedisLock(jedis,5000,requestId);
        System.out.println(reentrantRedisLock.lock("reentrantLock"));
        System.out.println(reentrantRedisLock.lock("reentrantLock"));
        System.out.println(reentrantRedisLock.unlock("reentrantLock"));
        System.out.println(reentrantRedisLock.unlock("reentrantLock"));
        System.out.println(reentrantRedisLock.unlock("reentrantLock"));
        RedisPool.release(jedis);
    }

    String topic = "t1";
    @Test
    public void producer() throws InterruptedException {
        Jedis jedis = RedisPool.getJedis();
        for(int i=0;i<10;i++){
            RedisProducer.send(jedis,topic,""+i);
        }
        RedisPool.release(jedis);
    }

    @Test
    public void consumer() throws InterruptedException {
        Jedis jedis = RedisPool.getJedis();

        try {
            RedisConsumer.recv(jedis,topic);
        }catch (Exception e){
            jedis = RedisPool.getJedis();
            RedisConsumer.recv(jedis,topic);
        }

        RedisPool.release(jedis);
    }

    @Test
    public void testScriptLoad(){
        Jedis jedis = RedisPool.getJedis();
        String luaScript = "local resultArray = redis.call('zrangebyscore', KEYS[1], 0, ARGV[1], 'limit' , 0, 1)\n" +
                "if #resultArray > 0 then\n" +
                "    if redis.call('zrem', KEYS[1], resultArray[1]) > 0 then\n" +
                "        return resultArray[1]\n" +
                "    else\n" +
                "        return ''\n" +
                "    end\n" +
                "else\n" +
                "    return ''\n" +
                "end";
        String sha1 = jedis.scriptLoad(luaScript);
        //5fafa096e6d8686d20b4915f22f1bdfa604dd965
        System.out.println(sha1);

        jedis.close();
    }


    @Test
    public void testStock() throws InterruptedException {
        Jedis jedis = RedisPool.getJedis();
        try {
            int stock = Integer.parseInt(jedis.get("stock"));
            System.out.println(stock);
        } finally {
            RedisPool.release(jedis);
        }
    }
    //监视一个值，事务执行过程中值不对，事务执行失败
    @Test
    public void testWatch(){
        Jedis jedis = RedisPool.getJedis();
        jedis.set("balance","100");
        jedis.watch("balance");//监视余额为100，改动了事务执行失败
//        jedis.set("balance","80");

        Transaction transaction = jedis.multi();
        transaction.set("balance","50");

        List<Object>result = transaction.exec();
        System.out.println(result);
        jedis.close();
    }
    //事务不是原子的，命令书写错误则整个事务执行失败，但是命令执行错误事务不会回滚，例如incr str
    //管道也不是原子的，批量操作，管道中某些命令执行失败也会返回错误给客户端，管道表达的语义是执行所有的命令
    //lua是原子的，redis确保这条script脚本执行期间，其它任何脚本或者命令都无法执行

    @Test
    public void testPipeline1() {
        Jedis jedis = RedisPool.getJedis();
        Pipeline pipelined = jedis.pipelined();
        String key = "k";

        pipelined.set(key,"v");
        pipelined.get(key);
        pipelined.incr(key);//异常
        List<Object> list = pipelined.syncAndReturnAll();
        for(Object obj :list){
            System.out.println(obj);
        }
        RedisPool.release(jedis);
    }

    @Test
    public void testPipeline() {
        Jedis jedis = RedisPool.getJedis();
        long begin = System.currentTimeMillis();
        for (int i = 1; i < 10000; i++) {
            jedis.get("stock");
        }
        System.out.println("get total time：" + (System.currentTimeMillis() - begin));

        Pipeline pipelined = jedis.pipelined();
        begin = System.currentTimeMillis();
        for (int i = 1; i < 10000; i++) {
            pipelined.get("stock");
        }
        //真正命令执行
        pipelined.sync();
        long end = System.currentTimeMillis();
        System.out.println("use pipeline batch set total time：" + (end - begin));
        jedis.close();
    }

    @Test
    public void testNormal() {
        Jedis jedis = RedisPool.getJedis();
        long begin = System.currentTimeMillis();
        for (int i = 1; i < 10000; i++) {
            jedis.get("stock");
        }
        jedis.close();
        long end = System.currentTimeMillis();
        System.out.println("no use pipeline batch set total time：" + (end - begin));
    }


    @Test
    public void testCluster() throws InterruptedException {
        JedisCluster jedisCluster = JedisClusterPool.getJedisCluster();
        msetV2(Arrays.asList("k","k1"),Arrays.asList("v","v1"));
        System.out.println(jedisCluster.get("k"));

        //mget error
//        System.out.println(jedisCluster.mget("k","k1"));
        System.out.println(mget("k", "k1"));
        System.out.println(keys("*"));
        del("*");

    }


    public void mset(List<String>keys,List<String>values){
        if(CollectionUtils.isEmpty(keys)||CollectionUtils.isEmpty(values)||keys.size()!= values.size()){
            return;
        }
        JedisCluster jedisCluster = JedisClusterPool.getJedisCluster();

        for(int i=0;i<keys.size();i++){
            String key = keys.get(i);
            String value = values.get(i);
            int slot = JedisClusterCRC16.getSlot(key);
            Jedis jedis = jedisCluster.getConnectionFromSlot(slot);
            jedis.set(key,value);
        }
    }

    public void msetV2(List<String>keys,List<String>values){
        if(CollectionUtils.isEmpty(keys)||CollectionUtils.isEmpty(values)||keys.size()!= values.size()){
            return;
        }
        JedisCluster jedisCluster = JedisClusterPool.getJedisCluster();

        Map<Jedis,List<Integer>> jedisToIndexs = new HashMap<>();

        for(int i=0;i<keys.size();i++){
            String key = keys.get(i);
            int slot = JedisClusterCRC16.getSlot(key);
            Jedis jedis = jedisCluster.getConnectionFromSlot(slot);

            int finalI = i;
            jedisToIndexs.compute(jedis,(jedis1, list)->{
                if(null==list){
                    list = new ArrayList<>();
                }
                list.add(finalI);
                return list;
            });
        }

        for(Map.Entry<Jedis,List<Integer>> entry:jedisToIndexs.entrySet()){
            Jedis jedis = entry.getKey();
            List<Integer> indexs = entry.getValue();
            String [] keyvalues = new String[indexs.size()*2];

            for(int i=0;i< indexs.size();i+=2){
                keyvalues[i] = keys.get(indexs.get(i));
                keyvalues[i+1] = values.get(indexs.get(i));
            }

            jedis.mset(keyvalues);
        }
    }

    public void del(String pattern){
        JedisCluster jedisCluster = JedisClusterPool.getJedisCluster();
        Map<String, JedisPool> nodes = jedisCluster.getClusterNodes();

        for(Map.Entry<String,JedisPool> entry:nodes.entrySet()){
            Jedis jedis = entry.getValue().getResource();
            if(!isMaster(jedis)){
                continue;
            }
            Pipeline pipeline = jedis.pipelined();
            String cursor = "0";
            //每次扫描10个
            ScanParams scanParams = new ScanParams().count(10).match(pattern);
            while(true){
                ScanResult<String> scanResult = jedis.scan(cursor, scanParams);
                List<String> keys = scanResult.getResult();
                if(CollectionUtils.isNotEmpty(keys)){
                    for (String key : keys) {
                        pipeline.del(key);
                    }
                    List<Object> objs = pipeline.syncAndReturnAll();
                    System.out.println(objs);
                }
                cursor = scanResult.getStringCursor();
                if("0".equals(cursor)){
                    break;
                }
            }
        }
    }


    public Map<String,String> mget(String... keys){
        List<String> list = new ArrayList<>();
        for (String key : keys) {
            list.add(key);
        }
       return mget(list);
    }



    public Set<String> keys(String pattern){
        JedisCluster jedisCluster = JedisClusterPool.getJedisCluster();
        Map<String, JedisPool> nodes = jedisCluster.getClusterNodes();
        Set<String> set = new HashSet<>();
        for(Map.Entry<String,JedisPool> entry:nodes.entrySet()){
            Jedis jedis = entry.getValue().getResource();
            if(isMaster(jedis)){
                set.addAll(jedis.keys(pattern));
            }
        }
        return  set;
    }

    private boolean isMaster(Jedis jedis) {
        String[] data = jedis.info("Replication").split("\r\n");
        for (String line : data) {
            if ("role:master".equals(line.trim())) {
                return true;
            }
        }
        return false;
    }

    //node -> keys    sdad
    public Map<String,String> mget(List<String> keys){
        JedisCluster jedisCluster = JedisClusterPool.getJedisCluster();

        Map<String,String> kvMap = new HashMap<>();

        Map<Jedis,List<String>> nodeToKeys = new HashMap<>();

        for (String key : keys) {
            int slot = JedisClusterCRC16.getSlot(key);
            Jedis jedis = jedisCluster.getConnectionFromSlot(slot);

            if(nodeToKeys.containsKey(jedis)){
                nodeToKeys.get(jedis).add(key);
            }else{
                List<String> list = new ArrayList<>();
                list.add(key);
                nodeToKeys.put(jedis,list);
            }
        }

        for(Map.Entry<Jedis,List<String>> entry: nodeToKeys.entrySet()){
            Jedis jedis = entry.getKey();

            List<String> nodeKeyList = entry.getValue();

            String[] nodeKeyArray = nodeKeyList.toArray(new String[nodeKeyList.size()]);

            List<String> nodeValueList = jedis.mget(nodeKeyArray);

            for (int i = 0; i < nodeKeyList.size(); i++) {
                kvMap.put(nodeKeyList.get(i), nodeValueList.get(i));
            }
        }
        return kvMap;
    }

    //redis秒杀 减库存
    //redis缓存一致性
    //redis缓存击穿 布隆过滤器
    //redis缓存雪崩

    public static void main(String[] args) {
        RedisClient client = RedisClient.create(RedisURI.create("redis://192.168.235.100:6379"));
        StatefulRedisConnection<String,String> connect = client.connect();

        /**
         * 同步调用
         */
        RedisCommands<String,String> commands = connect.sync();
        String str = commands.get("stock");
        System.out.println(str);

        /**
         * 异步调用
         */
        RedisAsyncCommands<String,String> asyncCommands = connect.async();
        RedisFuture<String> future = asyncCommands.get("stock");
        try {
            String str1 = future.get();
            System.out.println(str1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        connect.close();
        client.shutdown();
    }
}
